package commons.base.http;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.http.client.fluent.Content;
import org.apache.http.client.fluent.Request;
import org.apache.http.client.fluent.Response;
import org.apache.http.concurrent.FutureCallback;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;


/**
 * 有序HTTP请求运行器. 
 * 保证HTTP请求被有序的发送, 但是不保证响应是有序的
 * @author yuan<cihang.yuan@happyelements.com>
 *
 */
public class OrderHttpRequestExecutor {

	/**
	 *有序HTTP管道数
	 */
	public static final int DEFAULT_ORDER_HTTP_PIPE_COUNT = Runtime.getRuntime().availableProcessors();
	/**有序HTTP管道列表*/
	private final List<OrderHttpPipe> orderHttpPipeList = new ArrayList<OrderHttpPipe>();
	/**HTTP响应处理服务*/
	private final ExecutorService responseService = Executors.newFixedThreadPool(2);
	
	public OrderHttpRequestExecutor(){
		this(DEFAULT_ORDER_HTTP_PIPE_COUNT);
	}
	
	public OrderHttpRequestExecutor(int pipeCount){
		for(int i=0; i<pipeCount; i++){
			OrderHttpPipe orderHttpPipe = new OrderHttpPipe(responseService);
			orderHttpPipe.start();
			orderHttpPipeList.add(orderHttpPipe);
		}
	}
	
	/**
	 * 提交有序HTTP请求
	 * @param id
	 * @param request
	 * @param callback
	 */
	public void execute(long id, Request request, FutureCallback<Content> callback){
		int remainer = (int)(id % orderHttpPipeList.size());
		OrderHttpPipe orderHttpPipe = orderHttpPipeList.get(remainer);
		orderHttpPipe.writeRequest(request, callback);
	}
	
	public void shutdown(){
		for(OrderHttpPipe pipe : orderHttpPipeList){
			pipe.shutdown();
		}
		responseService.shutdown();
	}
}

/**
 * 有序的HTTP管道
 * 每个管道都是一个线程来处理请求有序的发送
 * 内部有一个阻塞队列保存等待发送的HTTP请求
 * @author yuan<cihang.yuan@happyelements.com>
 *
 */
final class OrderHttpPipe extends Thread{
	
	private static final Logger LOG = LogManager.getLogger(OrderHttpPipe.class);
	/**请求队列*/		
	private final BlockingQueue<OrderHttpRequest> queue = new LinkedBlockingQueue<OrderHttpRequest>();
	/**管道停止标志*/
	private AtomicBoolean stopping = new AtomicBoolean(false);
	/**HTTP响应处理服务*/
	private ExecutorService responseService;
	/**HTTP管道自动计数*/
	private static AtomicInteger pipeNumber = new AtomicInteger(1);
	
	public OrderHttpPipe(ExecutorService responseService) {
		super("OrderHttpPipe-" + pipeNumber.getAndIncrement());
		this.responseService = responseService;
	}
	
	/**
	 * 向管道写入HTTP请求
	 * @param request
	 * @param callback
	 */
	public void writeRequest(Request request, FutureCallback<Content> callback) {
		try {
			queue.put(new OrderHttpRequest(request, callback));
		} catch (InterruptedException e) {
			LOG.warn(e.getMessage(), e);
		}
	}
	
	/**
	 * 关闭管道
	 */
	public void shutdown(){
		stopping.set(true);
		interrupt();
	}
	
	public void run(){
		while (!stopping.get()) {
			try {
				final OrderHttpRequest orderHttpRequest = queue.take();
				try {
					Response response = orderHttpRequest.getRequest().execute();
					final Content content = response.returnContent();
					//成功
					responseService.submit(new Runnable() {
						public void run() {
							orderHttpRequest.getCallback().completed(content);
						}
					});
				} catch (final Exception e) {
					//失败
					responseService.submit(new Runnable() {
						public void run() {
							orderHttpRequest.getCallback().failed(e);
						}
					});
				}
			} catch (InterruptedException e) {
				LOG.warn(e.getMessage(), e);
			}
		}
	}
	
}

/**
 * 有序HTTP请求包装器
 * @author yuan<cihang.yuan@happyelements.com>
 *
 */
final class OrderHttpRequest{
	private Request request;
	private FutureCallback<Content> callback;
	
	public OrderHttpRequest(Request request, FutureCallback<Content> callback) {
		super();
		this.request = request;
		this.callback = callback;
	}
	public Request getRequest() {
		return request;
	}
	public void setRequest(Request request) {
		this.request = request;
	}
	public FutureCallback<Content> getCallback() {
		return callback;
	}
	public void setCallback(FutureCallback<Content> callback) {
		this.callback = callback;
	}
	
}